package com.gzsxy.esjy.bus.message.impl;


import com.alibaba.fastjson.JSON;
import com.gzsxy.esjy.bus.message.InnerProtocol;
import com.gzsxy.esjy.bus.message.MessageFactory;
import com.gzsxy.esjy.bus.subscrib.DefaultStream;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.binding.BinderAwareChannelResolver;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.MimeType;
import org.springframework.util.MimeTypeUtils;


/**
 * @author xiaolong
 * @version 1.0
 * @description: TODO
 * @date 2021/10/12 10:12
 */
@Slf4j
public class MessageFactoryImpl implements MessageFactory<InnerProtocol> {

    private final BinderAwareChannelResolver resolver;

    private final static String PREFIX = "/";

    public MessageFactoryImpl(BinderAwareChannelResolver resolver) {
        super();
        this.resolver = resolver;
    }


    @Value("${spring.application.name}")
    private String application;

    @Value("${spring.cloud.stream.bindings.default-in-broadcast.destination}")
    private String broadcast;


    @Override
    public void send(String dest, InnerProtocol message) {
        send(dest, message, MimeTypeUtils.APPLICATION_JSON);
    }

    @Override
    public void send(String dest, InnerProtocol message, MimeType type) {
        send(dest, message, type, -1);
    }


    @Override
    public void broadcast(InnerProtocol message) {

    }

    @Override
    public void broadcast(InnerProtocol message, MimeType type) {

    }

    @Override
    public void broadcast(InnerProtocol message, MimeType type, long timeOut) {

    }


    @Override
    public void send(final String dest, final InnerProtocol message, final MimeType type, final long timeOut) {
        String header = application;
        if (dest.equals(broadcast)) {
            header = broadcast;
        }
        String cmd = message.getCmd();
        while (cmd.startsWith(PREFIX)) {
            cmd = cmd.substring(PREFIX.length());
        }
        message.setCmd(PREFIX + cmd);

        //判断是否开启事务
        if (TransactionSynchronizationManager.isActualTransactionActive()) {
            //事务提交后发送消息
            final String finalHeader = header;
            TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                @Override
                public void afterCommit() {
                    resolver.resolveDestination(dest).send(MessageBuilder
                            .withPayload(message)
                            .setHeader(MessageHeaders.CONTENT_TYPE, type)
                            .setHeader(DefaultStream.HEADER_NAME, finalHeader)
                            .build(), timeOut);
                    log.info("消息发送dest={},内容={}", dest, JSON.toJSONString(message));
                }
            });

        } else {
            //直接发送消息  dest可以是通道名 也可以是主题名
            resolver.resolveDestination(dest).send(MessageBuilder
                    .withPayload(message)
                    .setHeader(MessageHeaders.CONTENT_TYPE, type)
                    .setHeader(DefaultStream.HEADER_NAME, header)
                    .build(), timeOut);
            log.info("消息发送dest={},内容={}", dest, JSON.toJSONString(message));
        }
    }


}
